#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <fcntl.h>
#include <pthread.h>
#include <errno.h>
#include <assert.h>

#include "base.h"
#include "conn.h"

static LIST_HEAD(freecq);
static pthread_mutex_t conn_lock = PTHREAD_MUTEX_INITIALIZER;

static int new_socket(void)
{
	int sfd;
	int flags;

	if ((sfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
		send_log(LV_ERROR, "#new_socket# socket() %s", strerror(errno));
		return -1;
	}

	if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 ||
			fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
		send_log(LV_ERROR, "#new_socket# fcntl() %s", strerror(errno));
		close(sfd);
		return -2;
	}
	return sfd;
}

int set_addr(const char * ip, int port, struct sockaddr_in * addr)
{
	assert(ip && port && addr);

	struct in_addr interface;
	interface.s_addr = htonl(INADDR_ANY);
	if (ip && !inet_pton(AF_INET, ip, &interface)) {
		send_log(LV_ERROR, "Illegal address: %s", ip);
		return -1;
	}

	/* 
	 * the memset call clears nonstandard fields in some impementations
	 * that otherwise mess things up.
	 */
	memset(addr, 0, sizeof(addr));

	addr->sin_family = AF_INET;
	addr->sin_port = htons(port);
	addr->sin_addr = interface;

	return 0;
}

int server_socket(const char * ip, int port, struct sockaddr_in * addr)
{
	assert(ip && port && addr);

	int sfd, flags = 1;
	struct linger ling = {0, 0};

	if ((sfd = new_socket()) < 0) {
		return -1;
	}

	if (set_addr(ip, port, addr) < 0) {
		return -2;
		close(sfd);
	}

	setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &flags, sizeof(flags));
	setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, &flags, sizeof(flags));
	setsockopt(sfd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling));

	if (bind(sfd, (struct sockaddr *)addr, sizeof(struct sockaddr_in)) == -1) {
		send_log(LV_ERROR, "#server_socket# bind() %s", strerror(errno));
		close(sfd);
		return -3;
	}

	if (listen(sfd, 1024) == -1) {
		send_log(LV_ERROR, "#server_socket# listen() %s", strerror(errno));
		close(sfd);
		return -4;
	}

	return sfd;
}


int client_socket(const char * ip, int port, struct sockaddr_in * addr)
{
	int sfd;

	if ((sfd = new_socket()) < 0) {
		return -1;
	}

	if (set_addr(ip, port, addr) < 0) {
		return -2;
		close(sfd);
	}

	if (connect(sfd, (struct sockaddr *)addr, sizeof(struct sockaddr_in)) < 0 && 
			errno != EINPROGRESS) {
		send_log(LV_ERROR, "#client_socket# connect() %s", strerror(errno));
		close(sfd);
		return -3;
	}

	return sfd;
}

int conn_accept(int sfd, struct sockaddr_in * addr)
{
	int newsfd, flags = 1;
	socklen_t addrlen = sizeof(struct sockaddr_in);
	if ((newsfd = accept(sfd, (struct sockaddr *)addr, &addrlen)) == -1) {
		if(errno == EAGAIN || errno == EWOULDBLOCK){
			return 0;
		}else{
			send_log(LV_ERROR, "#conn_accept# accept() %s", strerror(errno));
		}
		return -1;
	}
	if ((flags = fcntl(newsfd, F_GETFL, 0)) < 0 ||
			fcntl(newsfd, F_SETFL, flags | O_NONBLOCK) < 0) {
		send_log(LV_ERROR, "#conn_accept# fcntl() %s", strerror(errno));
		close(newsfd);
		return -2;
	}
	return newsfd;
}

static int wdata_free(WDATA * wdata)
{
	if (!wdata) {
		return 1;
	}
	free(wdata->wbuf);
	wdata->wbuf = NULL;
	wdata->wbytes = 0;
	free(wdata);
	return 0;
}

static WDATA * wdata_new(void * wbuf, int wbytes)
{
	assert(wbuf && wbytes > 0);
	WDATA * wdata = malloc(sizeof(struct WDATA));
	if (!wdata) {
		send_log(LV_FATAL, "#wdata_new# malloc() %s", strerror(errno));
		return NULL;
	}
	wdata->wbuf = wbuf;
	wdata->wbytes = wbytes;
	return wdata;
}

/* try handle packet
 * return
 * 0: no packet, ok
 * >0: handle one
 * <0: err
 */
static int try_handle_packet(CONN *c)
{
//	send_log(LV_DEBUG, "%s#try_handle_packet#", CTAG(c));
	if (c->rbytes == 0) return 0;

	int ret = 0;
	if (c->packlen == 0 &&
		c->depack(c->rcurr, c->rbytes, &c->headlen, &c->packlen, &c->scode) < 0) {
		send_log(LV_ERROR, "%s status code %d", CTAG(c), c->scode);
		c->state = conn_closing;
		return -1;
	} 

	if (c->packlen == 0 || c->packlen > c->rbytes) { //need to read more
		return 0;
	} else if (c->packlen < 0) {	//data error
		send_log(LV_ERROR, "%s illegal packlen %d", CTAG(c), c->packlen);
		c->state = conn_closing;
		return -2;
	}


	send_log(LV_DEBUG, "%d <= %d", c->packlen, c->rbytes);
	ret = c->onpack(c->rcurr, c->headlen, c->packlen, c->scode, c->obj);
	if (ret < 0) {
		send_log(LV_ERROR, "%s onpack ret = %d", CTAG(c), ret);
		c->state = conn_closing;
		return -3;
	}

	/* drop handled buffer */
	c->rcurr += c->packlen;
	c->rbytes -= c->packlen;

	c->headlen = c->packlen = c->scode = 0;

	return 1;
}

/* try read network 
 * return
 * 0: no data, ok
 * >0: read some
 * <0: err
 */
static int try_read_network(CONN *c)
{
	int gotdata = 0;
	int ret;

	send_log(LV_DEBUG, "%s #try_read_network#", CTAG(c));
	if (c->rcurr != c->rbuf) {
		if (c->rbytes != 0) /* otherwise there's nothing to copy */
			memmove(c->rbuf, c->rcurr, c->rbytes);
		c->rcurr = c->rbuf;
	}

	while (1) {
		if (c->rbytes >= c->rsize) {
			char *new_rbuf = realloc(c->rbuf, c->rsize*2);
			if (!new_rbuf) {
				send_log(LV_FATAL, "%s #try_read_network# realloc() %s", 
					CTAG(c), strerror(errno));
				c->rbytes = 0; /* ignore what we read */
				c->state = conn_closing;
				return -1;
			}
			c->rcurr  = c->rbuf = new_rbuf;
			c->rsize *= 2;
		}
		ret = read(c->sfd, c->rbuf + c->rbytes, c->rsize - c->rbytes);
		if (ret > 0) {
			gotdata = 1;
			c->rbytes += ret;
			send_log(LV_DEBUG, "%s #try_read_network# read %d/%d", CTAG(c), ret, c->rbytes);
			continue;
		}
		if (ret == 0) {
			/* connection closed */
			if (gotdata) {
				c->handle_and_go = conn_closing;
			} else {
				c->state = conn_closing;
			}
			return 1;
		}
		if (ret == -1) {
			if (errno == EAGAIN || errno == EWOULDBLOCK) break;
			else return -2;
		}
	}
	return gotdata;
}

static int try_write_network(CONN *c)
{
	int ret;
	if (c->wbytes == 0) {
		if (c->wdata) {
			//data free
			wdata_free(c->wdata);
			c->wdata = NULL;
		}
		if (list_empty(&c->wlist)) {
			c->state = c->write_and_go;
			return 0;
		}
		WDATA * wdata = list_entry(c->wlist.next, WDATA, list);
		assert(wdata && wdata->wbytes && wdata->wbuf);
		list_del(&wdata->list);
		c->wdata = wdata;
		c->wbytes = wdata->wbytes;
		c->wcurr = wdata->wbuf;
	}
	ret = write(c->sfd, c->wcurr, c->wbytes);
	if (ret > 0) {
		send_log(LV_DEBUG, "%s #try_write_network# write %d/%d [%s]", CTAG(c), ret, c->wbytes, c->wcurr);
		c->wcurr  += ret;
		c->wbytes -= ret;
	}
	return ret;
}

//only for debug
static char * cstats[] = {"listening", "connecting", "read", "write", "swallow", "closeing", NULL};

static int update_event(CONN *c, int new_flags);
static void drive_machine(CONN *c)
{
	int exit = 0;
	int sfd;
	struct sockaddr_in addr;
	CONN *newc;

	socklen_t optlen = sizeof(int);
	int ret = 0;

	send_log(LV_DEBUG, "%s -- drive --", CTAG(c));
	while (!exit) {
		send_log(LV_DEBUG, "%s continue, state %s", CTAG(c), cstats[c->state]);
		switch(c->state) {
			case conn_listening:
				if ((sfd = conn_accept(c->sfd, &addr)) <= 0) {
					if (sfd == 0) exit = 1;//again
					else send_log(LV_ERROR, "%s #drive# conn_accept", CTAG(c));
					break;
				}
				
				
				newc = conn_new(sfd, &addr, conn_read, EV_READ | EV_PERSIST, 
								c->base, c->depack, c->onpack, c->obj);
				if (!newc) {
					send_log(LV_ERROR, "%s #drive# create new connection", CTAG(c));
					close(sfd);
					break;
				}

				break;

			case conn_connecting:
				if (getsockopt(c->sfd, SOL_SOCKET, SO_ERROR, &ret, &optlen) < 0 ||
						ret != 0) {
					send_log(LV_ERROR, "%s #drive# connect getsockopt() %s | %s",
								CTAG(c), strerror(errno), strerror(ret));
					c->state = conn_closing;
					break;
				}
				c->state = conn_write;
				break;

			case conn_read:
				if (try_handle_packet(c)) {//sth handled
					continue;
				}
				if (c->handle_and_go != conn_read){
					c->state = c->handle_and_go;
					send_log(LV_DEBUG, "%s #drive# go close", CTAG(c));
					break;
				}
				if (try_read_network(c)) {//sth readed
					continue;
				}
				/* we have no command line and no data to read from network */
				if (update_event(c, EV_READ | EV_PERSIST) < 0) {
					send_log(LV_ERROR, "%s #drive# read update event", CTAG(c));
					c->state = conn_closing;
					break;
				}
				exit = 1;//exit means expect blocking
				break;

			case conn_write:
				/* we are writing wbytes bytes starting from wcurr */
				if (try_write_network(c) > 0) {
					continue;
				}

				if (update_event(c, EV_WRITE | EV_PERSIST) < 0) {
					send_log(LV_ERROR, "%s #drive# write update event", CTAG(c));
					c->state = conn_closing;
					break;
				}
				exit = 1;//exit means expect blocking
				break;
				/* if res==0 or res==-1 and error is not EAGAIN or EWOULDBLOCK,
				   we have a real error, on which we close the connection */
				send_log(LV_ERROR, "%s #drive# write, and not due to blocking", CTAG(c));
				c->state = conn_closing;
				break;

			case conn_closing:
				conn_close(c);
				exit = 1;
				break;
		}
	}

	return;
}

static void conn_handler(int fd, short which, void *arg)
{
	CONN *c = (CONN *)arg;
	assert(c);

	c->which = which;

	/* sanity */
	if (fd != c->sfd) {
		send_log(LV_ERROR, "%s Catastrophic: event fd doesn't match conn fd!", CTAG(c));
		conn_close(c);
		return;
	}

	/* do as much I/O as possible until we block */
	drive_machine(c);

	/* wait for next event */
	return;
}

static int update_event(CONN *c, int new_flags)
{
	struct event_base * base = c->event.ev_base;
	if (c->ev_flags == new_flags) return 0;

	if (event_del(&c->event) == -1) return -1;

	event_set(&c->event, c->sfd, new_flags, conn_handler, (void *)c);
	event_base_set(base, &c->event);
	c->ev_flags = new_flags;

	if (event_add(&c->event, 0) == -1) return -2;
	return 0;
}

CONN * conn_new(int sfd, struct sockaddr_in * addr, int init_state, int event_flags, 
		struct event_base *base, DEPACK_FUNC depack, ONPACK_FUNC onpack, const void *obj)
{
	CONN * c;

	/* do we have a free conn structure from a previous close? */
	if (!list_empty(&freecq)) {
		pthread_mutex_lock(&conn_lock);
		c = list_entry(freecq.next, CONN, list);
		list_del(&c->list);
		pthread_mutex_unlock(&conn_lock);
	} else { /* allocate a new one */
		if (!(c = malloc(sizeof(struct CONN)))) {
			send_log(LV_FATAL, "#conn_new# malloc() %s", strerror(errno));
			return NULL;
		}

		c->wdata = NULL;

		if (!(c->rbuf = malloc(DATA_BUFFER_SIZE))) {
			free(c);
			send_log(LV_FATAL, "#conn_new# malloc() %s", strerror(errno));
			return NULL;
		}
		c->rsize = DATA_BUFFER_SIZE;
		c->rcurr = c->rbuf;
	}


	c->sfd = sfd;
	c->state = init_state;
	c->rbytes = c->wbytes = 0;

	c->handle_and_go = conn_read;
	c->write_and_go = conn_read;
	INIT_LIST_HEAD(&c->wlist);

	c->headlen = c->packlen = c->scode = 0;

	memcpy(&c->addr, addr, sizeof(struct sockaddr_in));
	c->base = base;
	c->depack = depack;
	c->onpack = onpack;
	c->obj = obj;

	char ip[IPADDR_SIZE];
	inet_ntop(AF_INET, &addr->sin_addr, ip, IPADDR_SIZE - 1);
	snprintf(c->tag, TAG_SIZE - 1, "[<>%s:%d(%d)]", ip, ntohs(addr->sin_port), sfd);

	if (init_state == conn_listening) {
		send_log(LV_INFO, "*listen* %s", CTAG(c));
	} else if (init_state == conn_connecting) {
		c->tag[1] = '>';
		send_log(LV_INFO, "*connect* %s", CTAG(c));
	} else {
		c->tag[2] = '<';
		send_log(LV_INFO, "*accept* %s", CTAG(c));
	}

	event_set(&c->event, sfd, event_flags, conn_handler, (void *)c);
	event_base_set(c->base, &c->event);
	c->ev_flags = event_flags;

	if (event_add(&c->event, 0) == -1) {
		pthread_mutex_lock(&conn_lock);
		list_add_tail(&c->list, &freecq);
		pthread_mutex_unlock(&conn_lock);
		return NULL;
	}

	return c;
}

int conn_wlist_add(CONN * c, void * wbuf, int wbytes)
{
	assert(wbuf && wbytes);
	if (c->state == conn_closing) {
		send_log(LV_ERROR, "%s #wlist_add# invalid conn", CTAG(c));
		free(wbuf);
		return -1;
	}
	WDATA * wdata = wdata_new(wbuf, wbytes);
	if (!wdata) {
		free(wbuf);
		return -2;
	}
	list_add_tail(&wdata->list, &c->wlist);
	c->state = conn_write;
	if (update_event(c, EV_WRITE | EV_PERSIST) < 0) {
		wdata_free(wdata);
		c->state = conn_closing;
		send_log(LV_ERROR, "%s #wlist_add# write update event", CTAG(c));
		return -3;
	}
	return 0;
}

void conn_close(CONN *c)
{
	send_log(LV_INFO, "%s connection closing", CTAG(c));
	/* delete the event, the socket and the conn */
	if (event_del(&c->event) < 0) {
		send_log(LV_ERROR, "%s #conn_close# del event", CTAG(c));
	}

	close(c->sfd);

	while (!list_empty(&c->wlist)) {
		WDATA * wdata = list_entry(c->wlist.next, WDATA, list);
		list_del(&wdata->list);
		wdata_free(wdata);
	}

	pthread_mutex_lock(&conn_lock);
	list_add_tail(&c->list, &freecq);
	pthread_mutex_unlock(&conn_lock);

	return;
}

